use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{DestinationConnector, ConnectorConfig, ConnectorStatus, ConnectorMetrics};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use reqwest::Client;
use std::collections::HashMap;
use chrono::Utc;
use tracing::{info, error, warn};

/// HTTP目标连接器配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpDestinationConfig {
    /// 目标URL
    pub url: String,
    /// HTTP方法
    pub method: String,
    /// 请求头
    pub headers: HashMap<String, String>,
    /// 认证配置
    pub auth: Option<HttpAuthConfig>,
    /// 批次大小
    pub batch_size: u32,
    /// 请求超时时间（秒）
    pub timeout: Option<u64>,
    /// 重试配置
    pub retry: Option<RetryConfig>,
    /// 数据格式
    pub format: HttpDataFormat,
    /// 是否批量发送
    pub batch_mode: bool,
}

/// HTTP认证配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum HttpAuthConfig {
    /// Bearer Token认证
    Bearer { token: String },
    /// Basic认证
    Basic { username: String, password: String },
    /// API Key认证
    ApiKey { key: String, value: String, location: ApiKeyLocation },
}

/// API Key位置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ApiKeyLocation {
    Header,
    Query,
}

/// 重试配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryConfig {
    /// 最大重试次数
    pub max_retries: u32,
    /// 重试间隔（秒）
    pub retry_interval: u64,
    /// 指数退避
    pub exponential_backoff: bool,
}

/// HTTP数据格式
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum HttpDataFormat {
    /// JSON格式
    Json,
    /// JSON Lines格式
    JsonLines,
    /// 表单数据
    FormData,
    /// 纯文本
    Text,
}

impl ConnectorConfig for HttpDestinationConfig {
    fn validate(&self) -> Result<()> {
        if self.url.is_empty() {
            return Err(FluvioXError::Validation("URL不能为空".to_string()));
        }

        // 验证URL格式
        if !self.url.starts_with("http://") && !self.url.starts_with("https://") {
            return Err(FluvioXError::Validation("URL必须以http://或https://开头".to_string()));
        }

        // 验证HTTP方法
        match self.method.to_uppercase().as_str() {
            "GET" | "POST" | "PUT" | "DELETE" | "PATCH" => {},
            _ => return Err(FluvioXError::Validation("不支持的HTTP方法".to_string())),
        }

        if self.batch_size == 0 {
            return Err(FluvioXError::Validation("批次大小必须大于0".to_string()));
        }

        Ok(())
    }

    fn to_json(&self) -> Result<Value> {
        Ok(serde_json::to_value(self)?)
    }

    fn from_json(value: Value) -> Result<Self> {
        Ok(serde_json::from_value(value)?)
    }
}

/// HTTP目标连接器
pub struct HttpDestinationConnector {
    config: Option<HttpDestinationConfig>,
    client: Client,
    status: ConnectorStatus,
    metrics: ConnectorMetrics,
}

impl HttpDestinationConnector {
    pub fn new() -> Self {
        Self {
            config: None,
            client: Client::new(),
            status: ConnectorStatus::Uninitialized,
            metrics: ConnectorMetrics::default(),
        }
    }

    /// 构建HTTP请求
    async fn build_request(&self, config: &HttpDestinationConfig, data: &str) -> Result<reqwest::Request> {
        let mut request_builder = match config.method.to_uppercase().as_str() {
            "GET" => self.client.get(&config.url),
            "POST" => self.client.post(&config.url),
            "PUT" => self.client.put(&config.url),
            "DELETE" => self.client.delete(&config.url),
            "PATCH" => self.client.patch(&config.url),
            _ => return Err(FluvioXError::Validation("不支持的HTTP方法".to_string())),
        };

        // 添加请求头
        for (key, value) in &config.headers {
            request_builder = request_builder.header(key, value);
        }

        // 添加认证
        if let Some(auth) = &config.auth {
            request_builder = self.add_auth(request_builder, auth)?;
        }

        // 设置内容类型和请求体
        match config.format {
            HttpDataFormat::Json | HttpDataFormat::JsonLines => {
                request_builder = request_builder
                    .header("Content-Type", "application/json")
                    .body(data.to_string());
            },
            HttpDataFormat::FormData => {
                request_builder = request_builder
                    .header("Content-Type", "application/x-www-form-urlencoded")
                    .body(data.to_string());
            },
            HttpDataFormat::Text => {
                request_builder = request_builder
                    .header("Content-Type", "text/plain")
                    .body(data.to_string());
            },
        }

        // 设置超时
        if let Some(timeout) = config.timeout {
            request_builder = request_builder.timeout(std::time::Duration::from_secs(timeout));
        }

        request_builder.build()
            .map_err(|e| FluvioXError::Http(e.to_string()))
    }

    /// 添加认证信息
    fn add_auth(&self, mut builder: reqwest::RequestBuilder, auth: &HttpAuthConfig) -> Result<reqwest::RequestBuilder> {
        match auth {
            HttpAuthConfig::Bearer { token } => {
                builder = builder.bearer_auth(token);
            },
            HttpAuthConfig::Basic { username, password } => {
                builder = builder.basic_auth(username, Some(password));
            },
            HttpAuthConfig::ApiKey { key, value, location } => {
                match location {
                    ApiKeyLocation::Header => {
                        builder = builder.header(key, value);
                    },
                    ApiKeyLocation::Query => {
                        // 这里需要修改URL，简化处理
                        builder = builder.header(key, value);
                    },
                }
            },
        }

        Ok(builder)
    }

    /// 格式化记录数据
    fn format_records(&self, records: &[DataRecord], format: &HttpDataFormat) -> Result<String> {
        match format {
            HttpDataFormat::Json => {
                if records.len() == 1 {
                    Ok(serde_json::to_string(&records[0].data)?)
                } else {
                    let data: Vec<&Value> = records.iter().map(|r| &r.data).collect();
                    Ok(serde_json::to_string(&data)?)
                }
            },
            HttpDataFormat::JsonLines => {
                let mut lines = Vec::new();
                for record in records {
                    let line = serde_json::to_string(&record.data)?;
                    lines.push(line);
                }
                Ok(lines.join("\n"))
            },
            HttpDataFormat::Text => {
                let text = records.iter()
                    .map(|r| r.data.to_string())
                    .collect::<Vec<_>>()
                    .join("\n");
                Ok(text)
            },
            HttpDataFormat::FormData => {
                // 简化的表单数据格式化
                let mut form_data = Vec::new();
                for (i, record) in records.iter().enumerate() {
                    if let Value::Object(obj) = &record.data {
                        for (key, value) in obj {
                            form_data.push(format!("{}[{}]={}", key, i, value));
                        }
                    }
                }
                Ok(form_data.join("&"))
            },
        }
    }

    /// 发送HTTP请求（带重试）
    async fn send_with_retry(&self, config: &HttpDestinationConfig, data: &str) -> Result<bool> {
        let max_retries = config.retry.as_ref().map(|r| r.max_retries).unwrap_or(0);
        let retry_interval = config.retry.as_ref().map(|r| r.retry_interval).unwrap_or(1);
        let exponential_backoff = config.retry.as_ref().map(|r| r.exponential_backoff).unwrap_or(false);

        for attempt in 0..=max_retries {
            let request = self.build_request(config, data).await?;

            match self.client.execute(request).await {
                Ok(response) => {
                    if response.status().is_success() {
                        info!("HTTP请求成功，状态码: {}", response.status());
                        return Ok(true);
                    } else {
                        warn!("HTTP请求失败，状态码: {}", response.status());
                        if attempt == max_retries {
                            return Ok(false);
                        }
                    }
                },
                Err(e) => {
                    error!("HTTP请求执行失败: {}", e);
                    if attempt == max_retries {
                        return Err(FluvioXError::Http(e.to_string()));
                    }
                }
            }

            // 等待重试
            if attempt < max_retries {
                let wait_time = if exponential_backoff {
                    retry_interval * (2_u64.pow(attempt))
                } else {
                    retry_interval
                };

                info!("等待 {} 秒后重试...", wait_time);
                tokio::time::sleep(std::time::Duration::from_secs(wait_time)).await;
            }
        }

        Ok(false)
    }
}

#[async_trait]
impl DestinationConnector for HttpDestinationConnector {
    type Config = HttpDestinationConfig;

    fn name(&self) -> &str {
        "http-destination"
    }

    fn version(&self) -> &str {
        "1.0.0"
    }

    fn description(&self) -> &str {
        "HTTP目标连接器，支持多种认证方式和数据格式"
    }

    async fn initialize(&mut self, config: Self::Config) -> Result<()> {
        info!("初始化HTTP目标连接器");

        // 验证配置
        config.validate()?;

        self.config = Some(config);
        self.status = ConnectorStatus::Initialized;

        info!("HTTP目标连接器初始化完成");
        Ok(())
    }

    async fn test_connection(&self) -> Result<bool> {
        info!("测试HTTP连接");

        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;

        // 发送测试请求（空数据）
        match self.send_with_retry(config, "{}").await {
            Ok(success) => {
                if success {
                    info!("HTTP连接测试成功");
                } else {
                    warn!("HTTP连接测试失败");
                }
                Ok(success)
            },
            Err(e) => {
                error!("HTTP连接测试失败: {}", e);
                Ok(false)
            }
        }
    }

    async fn start(&mut self) -> Result<()> {
        info!("启动HTTP目标连接器");
        self.status = ConnectorStatus::Running;
        Ok(())
    }

    async fn stop(&mut self) -> Result<()> {
        info!("停止HTTP目标连接器");
        self.status = ConnectorStatus::Stopped;
        Ok(())
    }

    async fn write_records(&mut self, records: Vec<DataRecord>) -> Result<usize> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;

        if records.is_empty() {
            return Ok(0);
        }

        info!("发送 {} 条记录到HTTP端点", records.len());

        let mut total_sent = 0;
        let batch_size = config.batch_size as usize;

        // 分批处理
        for chunk in records.chunks(batch_size) {
            let formatted_data = self.format_records(chunk, &config.format)?;

            match self.send_with_retry(config, &formatted_data).await {
                Ok(true) => {
                    total_sent += chunk.len();
                    info!("成功发送 {} 条记录", chunk.len());
                },
                Ok(false) => {
                    error!("发送记录失败");
                    self.metrics.errors_count += 1;
                },
                Err(e) => {
                    error!("发送记录时出错: {}", e);
                    self.metrics.errors_count += 1;
                    return Err(e);
                }
            }
        }

        // 更新指标
        self.metrics.records_processed += total_sent as u64;
        self.metrics.last_activity = Some(Utc::now());

        info!("成功发送 {} 条记录", total_sent);
        Ok(total_sent)
    }

    fn status(&self) -> ConnectorStatus {
        self.status.clone()
    }

    fn metrics(&self) -> ConnectorMetrics {
        self.metrics.clone()
    }
}
